/**
 * 
 */
package edu.umd.clip.lm.storage;

import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class RequestBundle<K,T> implements AsyncConsumer<K, T> {
	private Lock lock;
	private Condition allDone;
	private Map<K,Long> requested;
	private Map<K,T> received;
	private final AsyncProducer<K,T> backend;
	private final static AtomicLong syncTime = new AtomicLong();
	private final static AtomicLong responseDelayTime = new AtomicLong();
	private final static AtomicLong syncCount = new AtomicLong();
	
	public RequestBundle(AsyncProducer<K,T> backend) {
		this.backend = backend;
		
		lock = new ReentrantLock();
		allDone = lock.newCondition();
		
		requested = new HashMap<K,Long>();
		received = new HashMap<K,T>();
	}
	
	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AsyncConsumer#receivedData(java.lang.Object, java.lang.Object)
	 */
	@Override
	public void receivedData(K key, T value) {
		lock.lock();
		received.put(key, value);
		long startTime = requested.get(key);
		long endTime = System.nanoTime();
		responseDelayTime.addAndGet(endTime - startTime);
		
		if (isDone()) {
			allDone.signalAll();
		}
		lock.unlock();
	}

	private boolean isDone() {
		return received.size() == requested.size();
	}
	
	public void sync() throws InterruptedException {
		long start = System.nanoTime();
		lock.lock();
		while(!isDone()) {
			long timeRemained = allDone.awaitNanos(8000000000l);
			if (timeRemained <= 0) {
				System.err.println("waiting for too long...");
			}
		}
		syncCount.incrementAndGet();
		lock.unlock();
		long end = System.nanoTime();
		syncTime.addAndGet(end - start);
	}

	public void request(K key) {
		lock.lock();
		boolean isNew = !requested.containsKey(key);
		if (isNew) {
			requested.put(key, System.nanoTime());
		}
		lock.unlock();
		if (isNew) {
			backend.request(key, this);
		}
	}
	
	public Map<K,T> getResults() {
		return received;
	}
	
	public static long getSyncTime() {
		return syncTime.get();
	}
	
	public static long getResponseDelayTime() {
		return responseDelayTime.get();
	}
	
	public static long getSyncCount() {
		return syncCount.longValue();
	}
}
